package com.lzq.mq.broker.netty.nameserver;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.broker.cache.CommonCache;
import com.lzq.mq.broker.config.GlobalProperties;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.coder.TcpMsgDecoder;
import com.lzq.mq.common.coder.TcpMsgEncoder;
import com.lzq.mq.common.constants.TcpConstants;
import com.lzq.mq.common.dto.PullBrokerIpReqDTO;
import com.lzq.mq.common.dto.PullBrokerIpRespDTO;
import com.lzq.mq.common.dto.ServiceRegistryReqDTO;
import com.lzq.mq.common.enums.*;
import com.lzq.mq.common.remote.NameServerNettyRemoteClient;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.util.internal.StringUtil;
import lombok.Getter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.net.Inet4Address;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import static com.lzq.mq.common.constants.BrokerConstants.*;

/**
 * 服务与nameserver服务器创建长链接, 支持链接创建, 支持重试机制
 */
@Slf4j
public class NameServerClient {

    @Getter
    private NameServerNettyRemoteClient nameServerNettyRemoteClient;

    private String brokerIp;

    private Integer brokerPort;

    /**
     * 初始化连接
     */
    public void initConnect() {
        // 创建nameserver客户端并连接nameserver
        String nameServerIp = CommonCache.getGlobalProperties().getNameServerIp();
        Integer nameServerPort = CommonCache.getGlobalProperties().getNameServerPort();
        nameServerNettyRemoteClient = new NameServerNettyRemoteClient(nameServerIp, nameServerPort);
        nameServerNettyRemoteClient.buildConnection();
        /*
        设置当前broker主服务的ip和端口
         */
        brokerPort = CommonCache.getGlobalProperties().getBrokerPort();

        if (nameServerIp.equals("127.0.0.1")) {
            brokerIp = "127.0.0.1";
        } else {
            try {
                brokerIp = Inet4Address.getLocalHost().getHostAddress();
            } catch (UnknownHostException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 发送注册请求
     */
    public void sendRegistryMsg() {

        try {

            // 获取当前程序的端口
            ServiceRegistryReqDTO registryDTO = new ServiceRegistryReqDTO();
            registryDTO.setIp(brokerIp);
            registryDTO.setPort(brokerPort);
            registryDTO.setUser(CommonCache.getGlobalProperties().getNameServerUser());
            registryDTO.setPassword(CommonCache.getGlobalProperties().getNameServerPassword());
            registryDTO.setRegistryType(RegistryTypeEnum.BROKER.getCode());
            registryDTO.setMsgId(UUID.randomUUID().toString());

            // broker是主从架构, 这里需要通过attr附加属性进行判断
            Map<String, Object> attrs = new HashMap<>();
            GlobalProperties globalProperties = CommonCache.getGlobalProperties();
            String clusterMode = globalProperties.getBrokerClusterMode();
            /*
              如果是单机模式
             */
            attrs.put(BROKER_GROUP_NAME_KEY, globalProperties.getBrokerClusterGroup());
            if(StringUtil.isNullOrEmpty(clusterMode) ||
                    globalProperties.getBrokerClusterRole().equals(BrokerRegistryRoleEnum.SINGLE.getCode())) {
                attrs.put(BROKER_ROLE_KEY, BrokerRegistryRoleEnum.SINGLE.getCode());
                globalProperties.setBrokerClusterRole(BrokerRegistryRoleEnum.SINGLE.getCode());
            } else if(BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(clusterMode)
                    && BrokerRegistryRoleEnum.MASTER.getCode().equals(globalProperties.getBrokerClusterRole())){
                //注册模式是集群架构
                attrs.put(BROKER_ROLE_KEY, BrokerRegistryRoleEnum.MASTER.getCode());
            } else if(BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(clusterMode)
                    && BrokerRegistryRoleEnum.SLAVE.getCode().equals(globalProperties.getBrokerClusterRole())) {
                attrs.put(BROKER_ROLE_KEY, BrokerRegistryRoleEnum.SLAVE.getCode());
            } else {
                log.error("注册模式错误:{}", globalProperties);
                throw new RuntimeException("注册模式错误!");
            }
            String msgId = registryDTO.getMsgId();
            registryDTO.setAttrs(attrs);
            registryDTO.setMsgId(msgId);
            byte[] body = JSON.toJSONBytes(registryDTO);
            TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.REGISTRY.getCode(), body);

            TcpMsg resp = nameServerNettyRemoteClient.sendSyncMsg(tcpMsg, msgId);

            int code = resp.getCode();
            if (code == NameServerResponseCode.REGISTRY_SERVICE_SUCCESS.getCode()) {
                // 注册成功
                log.info("服务注册成功");
            /*
                开启定时任务, 发送心跳包
             */
                CommonCache.getHeartBeatTaskManager().startTask();

            } else if (code == NameServerResponseCode.ERROR_USER_OR_PASSWORD.getCode()) {
                log.info("用户名或密码错误");
                throw new RuntimeException("用户名或密码错误");
            }
            log.info("注册broker为-{}", globalProperties.getBrokerClusterRole());
        } catch (Exception e) {
            log.error("发送注册请求失败", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取broker主节点的地址
     */
    public String queryBrokerMasterAddress() {
        PullBrokerIpReqDTO pullBrokerIpDTO = new PullBrokerIpReqDTO();
        pullBrokerIpDTO.setBrokerClusterGroup(CommonCache.getGlobalProperties().getBrokerClusterGroup());
        pullBrokerIpDTO.setRole(BrokerRegistryRoleEnum.MASTER.getCode());
        pullBrokerIpDTO.setMsgId(UUID.randomUUID().toString());
        TcpMsg tcpMsg  = new TcpMsg(NameServerEventCode.PULL_BROKER_IP_LIST.getCode(), JSON.toJSONBytes(pullBrokerIpDTO));
        TcpMsg pullBrokerIpResponse = nameServerNettyRemoteClient.sendSyncMsg(tcpMsg, pullBrokerIpDTO.getMsgId());
        PullBrokerIpRespDTO pullBrokerIpRespDTO = JSON.parseObject(pullBrokerIpResponse.getBody(),PullBrokerIpRespDTO.class);
        log.info("拉取broker主节点成功:{}", pullBrokerIpRespDTO);
        if (pullBrokerIpRespDTO == null || pullBrokerIpRespDTO.getAddressList() == null || pullBrokerIpRespDTO.getAddressList().isEmpty())
            return null;
        return pullBrokerIpRespDTO.getAddressList().get(0);
    }
}
